package mammuthus.template.hadoop

import mammuthus.template.{ServiceDiscover, TemplateProcessor}
import mammuthus.protocal.WriteFileTaskResponse
import mammuthus.tool.{Constants, TemplateTool}
import mammuthus.master.Master
import net.liftweb.{json => SJSon}
import net.sf.json.JSONArray
import net.csdn.common.logging.Loggers

/**
 * 5/5/15 WilliamZhu(allwefantasy@gmail.com)
 */
class HDFSSiteProcessor(master: Master) extends TemplateProcessor {

  val logger = Loggers.getLogger(classOf[HDFSSiteProcessor])

  case class NameNode(host: String, rpcPort: Int = 9000, httpPort: Int = 50070)
  implicit val formats = SJSon.Serialization.formats(SJSon.NoTypeHints)

  override def process(fileName: String, targetFilePath: String, params: Map[String, AnyRef]): List[WriteFileTaskResponse] = {
    val appName = params("appName").asInstanceOf[String]
    val app = params.getOrElse("app",appName)
    val zookeeper = params.getOrElse("zookeeper","zookeeper").asInstanceOf[String]
    val zkHosts = master.appInstallHosts(zookeeper)

    val slaves = params("slaves").asInstanceOf[String]
    val installPath = params.getOrElse("installPath",Constants.getAppRootDir)



    //必须有的参数
    val nameNodesStr = params("nameNodes").asInstanceOf[String]
    val nameNodes = nameNodesStr.split(",").take(2).map(f => NameNode(f))

    //我们默认是让qjm安装在所有namenode节点上。并且任意选出其中的两台，放在
    val res = master.writeJSONFile(slaves,Map(
     "nameNodes" -> nameNodes.map(f=>f.host).mkString(","),
     "qjm" -> nameNodesStr,
     "dataNodes" -> params("dataNodes").asInstanceOf[String]
    ), Constants.appInfoRootDir + appName, true)

    logger.info("写入系统:"+SJSon.Serialization.write(res))


    val qjm = nameNodesStr.split(",").map(f=>s"${f}:8485").mkString(";")
    val nameNodeMetaBaseDir = params.getOrElse("nameNodeDataBaseDir","/data/2").asInstanceOf[String]

    val slaveDataNodeMetaBaseDir = params.getOrElse("dataNodeDataBaseDir","/data/2").asInstanceOf[String].
      split(",").map(f => f + s"/${appName}/dfs/data").mkString(",")

    //可以提供默认的
    val dfsNameservices = params.getOrElse("dfs_nameservices","mammuthus")

    val templateParams = Map(
      "dfs_namenode_name_dir" -> s"$nameNodeMetaBaseDir/${appName}/dfs/name/fsimage",
      "dfs_namenode_edits_dir" -> s"$nameNodeMetaBaseDir/${appName}/dfs/name/edits",
      "dfs_namenode_shared_edits_dir" -> s"qjournal://${qjm}/${dfsNameservices}",
      "dfs_datanode_data_dir" -> slaveDataNodeMetaBaseDir,
      "dfs_nameservices" -> dfsNameservices,
      "dfs_ha_namenodes__" -> nameNodes.map(f => f.host).mkString(","),
      "dfs_namenodes" -> JSONArray.fromObject(SJSon.Serialization.write(nameNodes)),
      "dfs_zookeeper_nodes" -> zkHosts,
      "dfs_hosts_exclude" -> s"${installPath}/${appName}/etc/hadoop/dfs_exclude",
      "dfs_datanode_du_reserved" -> "107374182400",
      "dfs_balance_bandwidthPerSec" -> "3145728"
    )
    val configStr = TemplateTool.render(s"/template/app/$app/$fileName", templateParams)
    master.writeFile(slaves, configStr, targetFilePath, false)
  }
}
